package com.zf.nospring.kafka.transaction;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 事务消息，在一个事务中向 trans_test 主题发送消息，要么全都成功，要么全都失败
 */
public class TransactionProducerTest {
    public static void main(String[] args) {
        // 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // kafka broker的地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        // 设置事务id，如果想要开启事务，则必须设置该参数
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "ptrans");

        // 创建 kafka 生产者对象
        KafkaProducer<Object, Object> kafkaProducer = new KafkaProducer<>(properties);

        // 一旦开启了事务，就必须在事务内操作，在事务之外操作就报错，该行代码会报错
        // kafkaProducer.send(new ProducerRecord<>("test", "正常"));

        // 初始化事务
        kafkaProducer.initTransactions();
        // 开启事务, 开启了事务会默认开启幂等性
        kafkaProducer.beginTransaction();
        try {
            // 发送消息，由于开启了事务，所以只能在事务提交后，消费者才能消费这些消息。要么都发送成功，要么都发送失败
            // 不要忘记在消费者设置隔离级别为读取已提交的数据，否则可能会读取到未提交的数据
            for (int i = 1; i <= 5; i++) {
                kafkaProducer.send(new ProducerRecord<>("trans_test", "事务消息" + i)).get();
            }

            // 模拟异常
            // int a = 1/0;

            // 提交事务
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            e.printStackTrace();
            // 终止事务，类似于回滚
            kafkaProducer.abortTransaction();
        } finally {
            kafkaProducer.close();
        }
    }
}
